package com.burning.demo.flink.DataStreamAPI.demo1;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * Author: Mr.Deng
 * Date: 2018/10/15
 * Desc: 使用flink对指定窗口内的数据进行实时统计，最终把结果打印出来
 * 先在node21机器上执行nc -l -p 9999
 */
public class WindowWordCount {
    public static void main(String[] args) throws Exception {
	// 获取运行环境 - 使用非本地模式
	// StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
	// 使用本地模式并开启WebUI
	Configuration conf = new Configuration();
	conf.setString(RestOptions.BIND_PORT, "8081-8089");
	StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);
	// 定义socket的端口号
	// 连接socket获取输入的数据
	SingleOutputStreamOperator singleOutputStreamOperator = env.socketTextStream("localhost", 9999).flatMap(new Splitter())
		// 打平操作，把每行的单词转为<word,count>类型的数据
		// 针对相同的word数据进行分组
		.keyBy(value -> value.f0)
		// 指定计算数据的窗口大小和滑动窗口大小
		.window((WindowAssigner) TumblingProcessingTimeWindows.of(Time.seconds(10L))).sum(1);
	// 把数据打印到控制台,使用一个并行度
	singleOutputStreamOperator.print().setParallelism(1);

	// 注意：因为flink是懒加载的，所以必须调用execute方法，上面的代码才会执行
	env.execute("Window WordCount");
    }

    public static class Splitter implements FlatMapFunction<String, Tuple2<String, Integer>> {
	@Override
	public void flatMap(String sentence, Collector<Tuple2<String, Integer>> out) throws Exception {
	    byte b;
	    int i;
	    String[] arrayOfString;
	    for (i = (arrayOfString = sentence.split(" ")).length, b = 0; b < i;) {
		String word = arrayOfString[b];
		out.collect(new Tuple2(word, Integer.valueOf(1)));
		b++;
	    }
	}
    }
}
